{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "# 4. Callables in Python\n",
        "\n",
        "In Python, a **callable** is anything that can be “called” using `(...)`. This most commonly refers to functions (e.g., `def foo(): ...`), but **classes** (via `__call__`), **lambdas**, **functools.partial** objects, and so on can also be callables.\n",
        "\n",
        "## 4.1 Basic `Callable` Type Hints\n",
        "To specify that a variable, parameter, or attribute is a function or callable, Python provides the `Callable` type hint. The typical usage is:\n"
      ],
      "metadata": {
        "id": "cqIi0K_N_R4c"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from typing import Callable\n",
        "\n",
        "# A callable that takes two integers and returns a string\n",
        "MyFuncType = Callable[[int, int], str]\n",
        "\n",
        "print(MyFuncType)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "KDuig8I8AQfC",
        "outputId": "b841c6d1-35d9-4e31-cd9d-399df4366524"
      },
      "execution_count": 6,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "typing.Callable[[int, int], str]\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Usage\n",
        "from dataclasses import dataclass\n",
        "from typing import Callable\n",
        "\n",
        "@dataclass\n",
        "class Calculator:\n",
        "    operation: Callable[[int, int], str]\n",
        "\n",
        "    def calculate(self, a: int, b: int) -> str:\n",
        "        return self.operation(a, b)\n",
        "\n",
        "def add_and_stringify(x: int, y: int) -> str:\n",
        "    return str(x + y)\n",
        "\n",
        "calc = Calculator(operation=add_and_stringify)\n",
        "print(calc.calculate(5, 7))  # Outputs: '12'"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "qPj6ieXSDSXU",
        "outputId": "0f107122-8728-4f2a-a721-119ebc69b249"
      },
      "execution_count": 18,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "12\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 4.2 Generic Callables\n",
        "Generics let you parameterize a callable’s input or output types using `TypeVar`. For example:"
      ],
      "metadata": {
        "id": "IGigvUtQAYh8"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 24,
      "metadata": {
        "id": "JQPHf5ob_Nok"
      },
      "outputs": [],
      "source": [
        "from typing import Callable, TypeVar\n",
        "\n",
        "T = TypeVar(\"T\")\n",
        "U = TypeVar(\"U\")\n",
        "\n",
        "# A generic callable that transforms type T into type U\n",
        "Transformer = Callable[[T], U]"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "def apply_transformer(value: T, transformer: Callable[[T], U]) -> U:\n",
        "    return transformer(value)\n",
        "\n",
        "# Example 1: Transform an integer into a descriptive string.\n",
        "def int_to_str(n: int) -> str:\n",
        "    return f\"The number is {n}!\"\n",
        "\n",
        "result1 = apply_transformer(42, int_to_str)\n",
        "print(result1)  # Outputs: 'The number is 42!'"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "it2kkHxsGhUq",
        "outputId": "2febcc42-ea2f-4653-ecd8-ed67529f3c91"
      },
      "execution_count": 26,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "The number is 42!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "#### Usage with Generics and DataClass"
      ],
      "metadata": {
        "id": "kwBdgKymG0so"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "from dataclasses import dataclass\n",
        "from typing import Callable, Generic, TypeVar\n",
        "\n",
        "# TContext is a type variable used to parameterize our class.\n",
        "TContext = TypeVar(\"TContext\")\n",
        "\n",
        "@dataclass\n",
        "class PotionMixer(Generic[TContext]):\n",
        "    # The mix function now only expects a context of type TContext and returns a string.\n",
        "    mix: Callable[[TContext], str]\n",
        "\n",
        "    def create_potion(self, context: TContext) -> str:\n",
        "        return self.mix(context)\n",
        "\n",
        "# Example mixing function that uses a context (here, a dictionary) to create a potion description.\n",
        "def magical_mix(context: dict) -> str:\n",
        "    secret = context.get(\"secret\", \"moonlight\")\n",
        "    return f\"Potion of Wonder with a hint of {secret}!\"\n",
        "\n",
        "# Create an instance of PotionMixer with our magical_mix function.\n",
        "mixer = PotionMixer(mix=magical_mix)\n",
        "\n",
        "# Create a potion using a context that defines the secret ingredient.\n",
        "potion = mixer.create_potion({\"secret\": \"dragon scale\"})\n",
        "print(potion)  # Outputs: 'Potion of Wonder with a hint of dragon scale!'\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "7H2VYClXGY9B",
        "outputId": "861804e3-3e4e-4645-c18b-c061ae9caca4"
      },
      "execution_count": 23,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Potion of Wonder with a hint of dragon scale!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "\n",
        "## 4.3 Async vs. Sync Return Types with `MaybeAwaitable`\n",
        "**Sometimes,** we allow a callable to return **either** a normal (synchronous) value **or** an async `Awaitable` (such as a coroutine). A typical pattern is:"
      ],
      "metadata": {
        "id": "CYU-SIRSANPK"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import nest_asyncio\n",
        "nest_asyncio.apply()"
      ],
      "metadata": {
        "id": "qTz6oNe2J9Zr"
      },
      "execution_count": 40,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from collections.abc import Awaitable\n",
        "from typing import Callable, TypeVar, Union\n",
        "\n",
        "# Define a type variable for the input.\n",
        "T = TypeVar(\"T\")\n",
        "\n",
        "# A generic callable that takes a value of type T and returns something you may need to await to yield a string.\n",
        "MaybeAsyncFunc = Callable[[T], Union[Awaitable[str], str]]"
      ],
      "metadata": {
        "id": "sYO68aajAqBA"
      },
      "execution_count": 43,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import asyncio\n",
        "from collections.abc import Awaitable\n",
        "from typing import Callable, Union\n",
        "\n",
        "# A callable that takes a candidate's name (str) and returns a MaybeAwaitableStr.\n",
        "MaybeAsyncLeadAgent = Callable[[str], Union[Awaitable[str], str]]\n",
        "\n",
        "def lead_generation_agent(candidate: str) -> Union[Awaitable[str], str]:\n",
        "    \"\"\"\n",
        "    - If the candidate's name starts with a vowel, return a qualifying message immediately.\n",
        "    - Otherwise, simulate further review by returning an awaitable (a coroutine).\n",
        "    \"\"\"\n",
        "    if candidate[0].lower() in \"aeiou\":\n",
        "        return f\"Lead for {candidate}: Qualified immediately!\"\n",
        "    else:\n",
        "        async def review_lead() -> str:\n",
        "            return f\"Lead for {candidate}: Requires further review.\"\n",
        "        return review_lead()\n",
        "\n",
        "@dataclass\n",
        "class LeadGenerationAgent:\n",
        "  agent_launch_pad: MaybeAsyncLeadAgent\n",
        "\n",
        "  def __call__(self, candidate: str) -> Union[Awaitable[str], str]:\n",
        "    return self.agent_launch_pad(candidate)\n"
      ],
      "metadata": {
        "id": "U-Dp1UgoIeD3"
      },
      "execution_count": 44,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "agent = LeadGenerationAgent(lead_generation_agent)"
      ],
      "metadata": {
        "id": "SiB9KlENKSKu"
      },
      "execution_count": 46,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Case 1: Candidate whose name starts with a vowel; returns a string immediately.\n",
        "result1 = agent(\"Alice\")\n",
        "print(\"type\", type(result1))\n",
        "if asyncio.iscoroutine(result1):\n",
        "    result1 = asyncio.run(result1)\n",
        "print(result1)  # Expected output: \"Lead for Alice: Qualified immediately!\""
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "l95aGNXiJpxk",
        "outputId": "968fb260-0fac-420e-c92a-2d68f4dbb89a"
      },
      "execution_count": 47,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "type <class 'str'>\n",
            "Lead for Alice: Qualified immediately!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Case 2: Candidate whose name starts with a consonant; returns an awaitable.\n",
        "result2 = agent(\"Bob\")\n",
        "print(\"type\", type(result2))\n",
        "if asyncio.iscoroutine(result2):\n",
        "    result2 = asyncio.run(result2)\n",
        "print(result2)  # Expected output: \"Lead for Bob: Requires further review.\""
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "G3-wlKGQKRMa",
        "outputId": "03cd47e2-4c19-4bdc-e0fa-627164b362d8"
      },
      "execution_count": 48,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "type <class 'coroutine'>\n",
            "Lead for Bob: Requires further review.\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "---\n",
        "\n",
        "## 4.4 Combining Callables, Generics, and DataClasses\n",
        "You can embed a callable inside a DataClass, then pass it around in your code. This is powerful in AI agent architectures, where a callable might represent some “dynamic instructions” or “hook” that can be either sync or async.\n"
      ],
      "metadata": {
        "id": "yrtVcAFzAI2C"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import asyncio\n",
        "import inspect\n",
        "from dataclasses import dataclass\n",
        "from typing import Generic, TypeVar, Callable, Union, Optional\n",
        "from collections.abc import Awaitable\n",
        "\n",
        "TContext = TypeVar(\"TContext\")\n",
        "\n",
        "@dataclass\n",
        "class ContextWrapper(Generic[TContext]):\n",
        "    context: TContext\n",
        "\n",
        "# Define a type that can either be a static string or a callable that returns instructions.\n",
        "InstructionProvider = Union[\n",
        "    str,\n",
        "    Callable[[ContextWrapper[TContext]], Union[Awaitable[str], str]]\n",
        "]\n",
        "\n",
        "@dataclass\n",
        "class DynamicInstructions(Generic[TContext]):\n",
        "    instructions: InstructionProvider\n",
        "\n",
        "    def get_instructions(self, wrapper: Optional[ContextWrapper[TContext]] = None) -> Union[Awaitable[str], str]:\n",
        "        if callable(self.instructions):\n",
        "            if wrapper is None:\n",
        "                raise ValueError(\"A context must be provided for dynamic instructions\")\n",
        "            result = self.instructions(wrapper)\n",
        "            # If result is awaitable, ensure it's a coroutine\n",
        "            if inspect.isawaitable(result):\n",
        "                # If not a coroutine, wrap it in one.\n",
        "                if not inspect.iscoroutine(result):\n",
        "                    async def wrap():\n",
        "                        return await result\n",
        "                    result = wrap()\n",
        "                return asyncio.run(result)\n",
        "            return result\n",
        "        return self.instructions"
      ],
      "metadata": {
        "id": "FmiJACBo__Du"
      },
      "execution_count": 109,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "---\n"
      ],
      "metadata": {
        "id": "z08gmAgC_hCN"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Usage Example"
      ],
      "metadata": {
        "id": "he939BPO_6Em"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "#### a. Pass a string as instructions"
      ],
      "metadata": {
        "id": "Dvymbd25Ln9t"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# For static instructions, you don't need to pass any context.\n",
        "static_inst = DynamicInstructions[str](instructions=\"Write a 100 words tweet on Agentic AI in 2025.\")\n",
        "print(static_inst.get_instructions())"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "lxnr0l9RLq1f",
        "outputId": "a226aa9e-b47a-4298-fd55-104717b19f0f"
      },
      "execution_count": 110,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Write a 100 words tweet on Agentic AI in 2025.\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "#### b. Example with a sync callable"
      ],
      "metadata": {
        "id": "eiPLZGGLLnxr"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# For dynamic instructions, you'll need to provide a context.\n",
        "def sync_instruction_provider(ctx: ContextWrapper[str]) -> str:\n",
        "    return f\"Address me as: {ctx.context}\"\n",
        "\n",
        "sync_inst = DynamicInstructions[str](instructions=sync_instruction_provider)\n",
        "print(sync_inst.get_instructions(ContextWrapper(context=\"Junaid\")))"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "nfz1Wt4R_gYB",
        "outputId": "7ef74862-0afc-4172-9646-00d982c4cff6"
      },
      "execution_count": 111,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Address me as: Junaid\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "#### c. Example with an async callable"
      ],
      "metadata": {
        "id": "h5HUCO9_M8nT"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "async def async_instruction_provider(ctx: ContextWrapper[str]) -> str:\n",
        "    return f\"Address me as: {ctx.context}\"\n",
        "\n",
        "async_inst = DynamicInstructions[str](instructions=async_instruction_provider)\n",
        "\n",
        "import asyncio\n",
        "async_inst.get_instructions(ContextWrapper(context=\"Junaid\"))"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "id": "foAnSt0tMhri",
        "outputId": "ade25089-bf11-496a-a41c-90ae41a62dbd"
      },
      "execution_count": 112,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "'Address me as: Junaid'"
            ],
            "application/vnd.google.colaboratory.intrinsic+json": {
              "type": "string"
            }
          },
          "metadata": {},
          "execution_count": 112
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 4.5 Real-World Analogy for AI Agents\n",
        "In more advanced AI agent libraries (like in agents sdk), you often see fields like:\n",
        "\n",
        "```python\n",
        "instructions: (\n",
        "    str\n",
        "    | Callable[[RunContextWrapper[TContext], Agent[TContext]], MaybeAwaitable[str]]\n",
        "    | None\n",
        ") = None\n",
        "```\n",
        "\n",
        "This means:\n",
        "1. **`str`**: a simple, static prompt or instructions text.\n",
        "2. **`Callable[[RunContextWrapper[TContext], Agent[TContext]], MaybeAwaitable[str]]`**:  \n",
        "   a function taking two parameters (`RunContextWrapper` and the `Agent` itself) that can return either a `str` (sync) or an async “awaitable” string.\n",
        "3. **`None`**: no instructions at all.\n",
        "\n",
        "When the agent runs:\n",
        "- If `instructions` is a string, it’s used directly.\n",
        "- If it’s a callable, we call or `await` it to obtain the instructions dynamically.\n",
        "- If `None`, we might skip or throw an error.\n",
        "\n",
        "This design approach offers **maximal flexibility**: you can store a static string for simpler use cases, or pass a function that can look at the user’s context, model settings, or any other runtime data to compute a specialized prompt.\n",
        "\n",
        "---\n",
        "\n",
        "## Summary\n",
        "\n",
        "1. **`Callable` Type Hint**: The cornerstone for describing function signatures in typed Python code.\n",
        "2. **Generics with `Callable`**: Allows you to express “a function that transforms `T` into `U`,” capturing the function’s input and output types for strong type-checking.\n",
        "3. **`MaybeAwaitable`**: A powerful pattern that supports both synchronous and asynchronous return types, enabling flexible usage in complex AI or concurrency scenarios.\n",
        "4. **DataClasses + Callables**: You can store callables inside data classes to build more flexible, configurable systems—particularly in AI agent frameworks where instructions, hooks, or tool functions can be dynamic.\n",
        "\n",
        "By combining these patterns—**Generics**, **DataClasses**, **Protocols**, **Callables**, and **MaybeAwaitable**—we can create highly **type-safe**, **extensible**, and **readable** architectures that scale from simple prototypes to production-grade AI agent systems."
      ],
      "metadata": {
        "id": "vwrqlHx3_djL"
      }
    }
  ]
}